package io.sqrtqiezi.spark.lagou

import org.apache.spark.{SparkConf, SparkContext}

// 数据格式：时间点			省份		城市	用户		广告
object Adstat {

  def getHour(str: String) = {
    import org.joda.time.DateTime

    val dt = new DateTime(str.toLong)
    dt.getHourOfDay
  }

  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init)
      .setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val N = 3
    val lines = sc.textFile("lagou-data/advert.log")

    val stat1RDD = lines.map { line =>
      val fields = line.split("\\s+")
      (fields(1), fields(4))
    }

    // 按省份、广告汇总
    val reduce1RDD = stat1RDD
      .map { case (province, adid) => ((province, adid), 1) }
      .reduceByKey(_ + _)

    // 对汇总信息取 top3
    reduce1RDD.map{case ((province, adid), count) => (province, (adid, count))}
      .groupByKey()
      .mapValues(buf => buf.toList.sortWith(_._2 > _._2).take(3).map(_._1).mkString(":"))
      .foreach(println)

    // 2. 统计每一个省每一个小时的 Top3 广告ID
    lines.map {line =>
      val fields = line.split("\\s+")
      ((getHour(fields(0)), fields(1), fields(4)), 1)
    }.reduceByKey(_+_)
      .map{case ((hour, province, adid), count) =>
        ((province, hour), (adid, count))
      }
      .groupByKey()
      .mapValues(buf => buf.toList.sortWith(_._2 > _._2).take(3).map(_._1).mkString(":"))
      .collect.foreach(println)

    sc.stop()
  }
}
